/**   
 * @Title: QueueListenter.java 
 * @Package com.cyl.rabbitmq 
 * @Description: TODO(��һ�仰�������ļ���ʲô) 
 * @author zjhua@hundsun.com   
 * @date 2016��4��25�� ����1:15:31 
 * @version V1.0   
 */
package com.medsoft.logpool.core;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.Consumer;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;

/**
 * @author zjhua
 *
 */
@Service
public class LogQueueListener implements InitializingBean {
	protected static final Logger log = LoggerFactory
			.getLogger(LogQueueListener.class);

	@Autowired
	private ConnectionFactory connectionFactory;

	private Channel channel;

	@Autowired
	private LogDAO logDAO;
	/*
	 * (non-Javadoc)
	 * 
	 * @see
	 * org.springframework.amqp.core.MessageListener#onMessage(org.springframework
	 * .amqp.core.Message)
	 * 
	 */
	@Transactional
	public void onMessage(String msg) {
		
		Map<String, String> param = new HashMap<String,String>();
		String[] logRecord = msg.split("###");
		param.put("logTime", logRecord[0]);
		param.put("logLevel", logRecord[1]);
		param.put("hostname", logRecord[2].substring(0,Math.min(logRecord[2].length(), 64)));
		param.put("appName", logRecord[3].substring(0,Math.min(logRecord[3].length(), 64)));
		param.put("threadName", logRecord[4].substring(0,Math.min(logRecord[4].length(), 64)));
		param.put("source", logRecord[5].substring(0, Math.min(logRecord[5].length(), 256)));
		param.put("message_len", String.valueOf((logRecord[6] + "\r\n" + logRecord[7]).length()));
		int trunc = 0;
		if ((logRecord[6] + "\r\n" + logRecord[7]).length() <= 20000) {
			trunc = (logRecord[6] + "\r\n" + logRecord[7]).length();
		} else {
			trunc = 20000;
		}
		param.put("message", (logRecord[6] + "\r\n" + logRecord[7]).substring(0, trunc));
		try {
			logDAO.addLog(param);
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see
	 * org.springframework.beans.factory.InitializingBean#afterPropertiesSet()
	 */
	@Override
	public void afterPropertiesSet() throws Exception {
		Connection connection = connectionFactory.newConnection();

		channel = connection.createChannel();

		Consumer consumer = new DefaultConsumer(channel) {
			@Override
			public void handleDelivery(String consumerTag, Envelope envelope,
					AMQP.BasicProperties properties, byte[] body)
					throws IOException {
				String message = new String(body, "UTF-8");
				log.warn(message);
				onMessage(message);
			}
		};
		channel.basicConsume("service.log4j.events", true, consumer);
	}
}
